package com.lagou.niancheng.rpc.common.handler;

import com.lagou.niancheng.rpc.common.converter.Converter;
import com.lagou.niancheng.rpc.common.entity.SimpleRpcProtocol;

import com.lagou.niancheng.rpc.common.entity.SimpleRpcProtocolBuilder;
import com.lagou.niancheng.rpc.common.enums.SignalEnum;
import com.lagou.niancheng.rpc.common.exceptions.SimpleRpcException;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.util.Map;

import static com.lagou.niancheng.rpc.common.entity.SimpleRpcProtocolBuilder.DATA_HEADER_KEY;

/**
 * 将协议对象转成传输对象，并交给相应的请求处理器处理
 *
 * @author 大犬
 * @date 2020/4/30
 */
public class SimpleRpcTransferDataHandler extends SimpleChannelInboundHandler<SimpleRpcProtocol> {

    private Converter converter;

    private RequestHandlerRegistry requestHandlerRegistry;

    public SimpleRpcTransferDataHandler(Converter converter, RequestHandlerRegistry requestHandlerRegistry) {
        this.converter = converter;
        this.requestHandlerRegistry = requestHandlerRegistry;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, SimpleRpcProtocol simpleRpcProtocol) throws Exception {
        byte[] headerContent = simpleRpcProtocol.getHeader();
        byte[] dataContent = simpleRpcProtocol.getContent();
        Map header = (Map) converter.bytes2Obj(headerContent, Map.class);
        // 从请求头部中获取数据类型
        String dataTypeName = (String) header.get(DATA_HEADER_KEY);
        Class<?> dataType = Class.forName(dataTypeName);
        RequestHandler requestHandler = requestHandlerRegistry.getRequestHandler(dataType);
        if (requestHandler == null) {
            SimpleRpcProtocol errorProtocolData = SimpleRpcProtocolBuilder.buildErrorProtocolData(new SimpleRpcException("cannot handle this request"), converter);
            channelHandlerContext.channel().writeAndFlush(errorProtocolData);
            return;
        }
        Object o = converter.bytes2Obj(dataContent, dataType);
        @SuppressWarnings("unchecked")
        Object result = requestHandler.handle(o, channelHandlerContext.channel());
        if (SignalEnum.FINISH_CONNECTION.equals(result)) {
            // 连接关闭
            channelHandlerContext.disconnect().addListener(ChannelFutureListener.CLOSE);
            return;
        }
        SimpleRpcProtocol successProtocolData = SimpleRpcProtocolBuilder.buildSuccessProtocolData(result, converter);
        channelHandlerContext.channel().writeAndFlush(successProtocolData);
    }
}
